/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.tcp;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;

import cgl.narada.protocol.Protocol;
import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportHandler;

/**
 * Listens for TCP packets arriving at a socket and performs various operations
 * based on situations arising during communication. These received packets are
 * then encapsulated into a specialized data structure and forwarded to the
 * transport handler implenetation residing in the transport layer.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class TCPReceiverThread extends Thread implements TransportDebugFlags {

    private Socket socket;

    private DataInputStream din;

    private BufferedInputStream bin;

    private int packetTracker = 0;

    private boolean zeroLengthRead = false;

    private long readingDelayStart;

    private TCPLink tcpLink;

    private String moduleName = "TCPReceiverThread: ";

    private TransportHandler transportHandler;

    private int receiveBufferSize = 12 * 8192;

    private String linkId;

    private boolean shutdownService = false;
    
    public TCPReceiverThread(Socket socket) {
        initializeDataInputStream(socket);
    }


    public TCPReceiverThread(Socket socket, TCPLink tcpLink) {
        this.tcpLink = tcpLink;
        initializeDataInputStream(socket);
        transportHandler = tcpLink.getTransportHandler();
        if (transportHandler == null) {
            System.out.println(moduleName + "TransportHandler is null");
        }
    }


    private void initializeDataInputStream(Socket socket) {
        this.socket = socket;
        try {
            socket.setSoLinger(true, 2);
            socket.setTcpNoDelay(true);

            byte tc = 0x08;
            socket.setTrafficClass(tc);
            socket.setKeepAlive(true);
            socket.setReceiveBufferSize(receiveBufferSize);
            if (TCPReceiverThread_Debug) {
                System.out.println(moduleName + "Receive Buffer Size ="
                    + socket.getReceiveBufferSize() + ", traffic class = "
                    + socket.getTrafficClass() + ", TcpNODelay="
                    + socket.getTcpNoDelay());
            }

        } catch (SocketException se) {
            se.printStackTrace();
        }

        try {
            din = new DataInputStream(socket.getInputStream());
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
    }


    public void setLinkId(String linkId) {
        if (linkId == null) {
            System.out.println(moduleName + "Specified Link ID is null");
        }
        this.linkId = linkId;
    }


    public void releaseResources() {
        if (TCPReceiverThread_Debug)
            System.out.println(moduleName + "Closing " + socket);
        try {
            shutdownService = true;
            socket.getInputStream().close();
            socket.close();
        } catch (IOException ioe) {
            System.out.println(moduleName + "Problem while releasing resources"
                + ioe);
        }
    }


    public void run() {
        if (TCPReceiverThread_Debug)
            System.out.println("TCPReceiverThread listening for messages to "
                + socket);
        int dataLength;
        int packetTracker = 0;
        byte transportHeader;
        while (socket != null) {
            try {
                if (TCPReceiverThread_Debug)
                    System.out.println("Receiving ..");

                transportHeader = din.readByte();

                if (transportHeader == Protocol.PING) {
                    System.out.println("Received ping ");
                    // senderThread.sendPingResponse(din.readLong());
                    // Enabled by HG:
                    din.readLong();
                    tcpLink.sendPingReply();
                    continue;
                }

                if (transportHeader == Protocol.PING_REPLY) {
                    long timeSent = din.readLong();
                    System.out.println("Ping time = "
                        + (System.currentTimeMillis() - timeSent));
                    continue;
                }

                if (transportHeader != Protocol.DATA) {
                    System.out.println("\n \n" + moduleName
                        + "MAJOR ERROR: Received Data with garbled "
                        + "transport header" + "\n \n");
                    continue;
                }

                dataLength = din.readInt();
                if (dataLength > 1000000) {
                    if (TCPReceiverThread_Debug) {
                        System.out.println(moduleName + "Flagging data of size["
                            + dataLength + "] on link " + linkId);
                    }
                }
                if (TCPReceiverThread_Debug) {
                    System.out.println(moduleName + "Reading data of size ["
                        + dataLength + "] on link " + linkId);
                }

                // long startTime = System.currentTimeMillis();
                byte[] data = new byte[dataLength];
                din.readFully(data, 0, dataLength);
                // long elapsedTime = System.currentTimeMillis() - startTime;
                // String report = moduleName + "Time to read (" + dataLength +
                // ") = " +
                // elapsedTime;
                // System.out.println(report);

                packetTracker++;
                if (TCPReceiverThread_Debug) {
                    System.out.println(moduleName + "Read data of size ["
                        + dataLength + "] on link " + linkId);
                    System.out.println(moduleName + "Packet Number="
                        + packetTracker + " , length= " + dataLength);
                }

                transportHandler.dataReceived(data, linkId);

            } catch (SocketException se) {
                if (!shutdownService) {
                    System.out.println(moduleName + se);
                }
                break;
            } catch (IOException ioe) {
                System.out.println(moduleName + ioe);
                break;
            }
        }

        /** Report loss of connection to the TCP Link */
        tcpLink.processLossOfSocketConnection(false);

        if (TCPReceiverThread_Debug)
            System.out.println("TCPReceiver: Socket Connection to [" + socket
                + "] broken");

    }
}
